package com.youxin.logcenter.kafka;

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;

import com.youxin.busfoundation.config.KafkaConfig;
import com.youxin.busfoundation.enums.SystemType;
import com.youxin.foundation.utils.JsonUtils;
import com.youxin.foundation.utils.thread.ScheduleThreadPool;
import com.youxin.logcenter.dao.LogcpMapper;
import com.youxin.logcenter.dao.LogoaMapper;
import com.youxin.logcenter.dao.LogpcMapper;
import com.youxin.logcenter.model.Logcp;
import com.youxin.logcenter.model.Logoa;
import com.youxin.logcenter.model.Logpc;

/**
 * 消费日志
 * 
 * @author yingde.cao
 *
 */
@Service
public class KafkaConsumerLog {
	private static Logger logger = LoggerFactory.getLogger(KafkaConsumerLog.class);
	
	@Autowired 
	private LogpcMapper logpcMapper;
	@Autowired 
	private LogoaMapper logoaMapper;
	@Autowired 
	private LogcpMapper logcpMapper;

	public void start() {
		KafkaConfig config=KafkaConfig.getConfig();
		KafkaConsumer<String, String> consumer = null;
		try {
			Properties properties = new Properties();
			properties.put("bootstrap.servers", config.getServers());
			properties.put("group.id", config.getGroupId());
			properties.put("enable.auto.commit", "false");
			properties.put("auto.commit.interval.ms", "1000");
			properties.put("session.timeout.ms", "30000");
			properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
			properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
			properties.put("auto.offset.reset", "earliest");			
			
			consumer = new KafkaConsumer<>(properties);
			consumer.subscribe(Arrays.asList(config.getTopics().split(",", 0)));
			while (true) {
				ConsumerRecords<String, String> records = consumer.poll(100);
				for (ConsumerRecord<String, String> record : records) {
					try {
						String json=record.value();
						String key=record.key();				
						if (SystemType.PC.getName().equalsIgnoreCase(key))
						{			
							Logpc logBean=JsonUtils.stringToFasterxmlObject(json, Logpc.class);	
							int rowEffect=logpcMapper.insertSelective(logBean);
							if(rowEffect==1){
								consumer.commitAsync();
							}			
						}else if (SystemType.OA.getName().equalsIgnoreCase(key))
						{
							Logoa logBean=(Logoa)JsonUtils.stringToObject(json, Logoa.class);	
							int rowEffect=logoaMapper.insertSelective(logBean);
							if(rowEffect==1){
								consumer.commitAsync();
							}	
						}else if (SystemType.CP.getName().equalsIgnoreCase(key))
						{
							Logcp logBean=(Logcp)JsonUtils.stringToObject(json, Logcp.class);	
							int rowEffect=logcpMapper.insertSelective(logBean);
							if(rowEffect==1){
								consumer.commitAsync();
							}	
						}else{
							/**
							 * 不能识别的消息类型
							 */
							consumer.commitAsync();
						}
					}catch (DuplicateKeyException e) {
						try {
							consumer.commitAsync();
							logger.error("日志入库失败", e);
						} catch (Exception e2) {
						}
					}
					catch(DataIntegrityViolationException e){
						try {
							consumer.commitAsync();
							logger.error("日志入库失败", e);
						} catch (Exception e2) {
						}
					}
					catch (Exception e) {
						logger.error("日志入库失败", e);
					}
				}
			}
		} catch (Exception e) {
			logger.error("Kafka异常", e);
		} finally {
			try {
				if (consumer != null) {
					consumer.close();
				}
			} catch (Exception e) {
			}
		}

	}
	
	/**
	 * 执行方法
	 */
	public void execute() {
		try {
			ScheduleThreadPool.scheduleWithFixedDelay(new Runnable() {				
				@Override
				public void run() {
					start();					
				}
			}, 60000, 0);
			
		} catch (Exception e) {
			logger.error("Execute", e);
		}
	}
}
